<?php
namespace PWC;
class Agent{
    private $sites;
    private $total=0;
    private $config;
    private $lockid;
    private $receiver;
    private $config_file;

    private $agent_str = 'Mozilla/5.0 (Windows NT 5.1; rv:9.0) Gecko/20100101 Firefox/9.0';
    private $header    = array(
        # '-H "Accept-Encoding: gzip"', // 使用压缩，以减少带宽，但会增加CPU的负担
    );
    public function getTotalNumber(){
        return $this->total;
    }
    public function getConfig(){
        return $this->config;
    }
    public function initialize($config_file){
        $this->config_file = $config_file;
        $this->config = Common::xmlConfig($config_file);
        $this->db->exec(
            'CREATE TABLE IF NOT EXISTS `ongoing`(
                `type` ENUM("site","connect") NOT NULL, 
                `type_id` BINARY(16) NOT NULL,
                `num` SMALLINT UNSIGNED NOT NULL DEFAULT 0,
                `last_time` DECIMAL(15,4) UNSIGNED NOT NULL DEFAULT 0,
                UNIQUE (`type`,`type_id`)
            ) ENGINE=MEMORY'
        );
        $this->db->exec(
            'CREATE TABLE IF NOT EXISTS `urls`(
                `url` VARCHAR(500) NOT NULL UNIQUE,
                `ticket` INT UNSIGNED NOT NULL DEFAULT 0,
                `referer` VARCHAR(500) NULL,
                `last_time` INT UNSIGNED NOT NULL DEFAULT 0
            ) ENGINE=MEMORY'
        );
        $this->db->exec('DELETE FROM `ongoing` WHERE `num`=0');
        $this->sites = array();

        $urls_sql = 'INSERT IGNORE INTO `urls`(`url`,`last_time`) VALUES (?,1)';
        $urls_sth = $this->db->prepare($urls_sql);
        foreach($this->config->sites as $site){
            $urls_sth->execute(array($site['site']));
            $url_info = parse_url($site['site']);
            $this->sites[$url_info['host']] = $site;
            unset($this->sites[$url_info['host']]['site']);
        }

        $ongoing_sql = 'INSERT IGNORE INTO `ongoing`(`type`,`type_id`) VALUES (?,?)';
        $ongoing_sth = $this->db->prepare($ongoing_sql);
        foreach($this->config->connects as $connect){
            $ongoing_sth->execute(array('connect',$this->getSign($connect)));
            foreach($this->config->sites as $site){
                $ongoing_sth->execute(array('site',$this->getSign($connect,$site['site'])));
            }
        }
        #print_r($this);
    }
    private function getSign($connect, $url=''){
        $sign = $connect['type'].'://'.$connect['host'];
        if($url){
            if(!preg_match('#^[a-z]+://#',$url)){ // like www.abc.com/index.php
                $url = 'http://'.$url;
            }
            $site = parse_url($url);
            $sign .= '|'.$site['host'];
        }
        return md5($sign,1);
    }

    public function __construct($config_file){
        $config = Common::xmlConfig($config_file);
        $this->config_file = $config_file;
        $this->config = $config;

        $this->lockid  = sem_get(ftok(__FILE__,'r'), 1, 0644);
        $this->db = new \PDO(
            "mysql:dbname={$config->mysql['name']};host={$config->mysql['host']}",
            $config->mysql['user'],
            $config->mysql['pass']
        );

        foreach($this->config->sites as $site){
            $site_info = parse_url($site['site']);
            unset($site['site']);
            $this->sites[$site_info['host']] = $site;
        }
        $this->receiver = dirname(__FILE__).'/receiver.php';
    }
    public function remove(){
        sem_remove($this->lockid);
    }

    public function setCookie($url, $cookie_str){
        $now  = time();
        $old  = $new = array();
        $path = parse_url($url);
        $file = $this->config->cookie_store_path.'/'.$path['host'].'.cookie';

        #0:domain 1:flag 2:path 3:secure 4:expiration 5:name 6:value
        $lines = explode("\n",$cookie_str);
        foreach($lines as $line){
            $one = explode("\t",$line);
            if(count($one)!=7) continue;
            $new[$one[5]] = $one; 
        }

        sem_acquire($this->lockid);

        if(file_exists($file)){
            $lines = explode("\n",file_get_contents($file));
            foreach($lines as $line){
                $one = explode("\t",$line);
                $name = $one[5];
                if($one[4] > $now){ // No Expired
                    if(!array_key_exists($name,$new) || $one[0]!=$new[$name][0]){
                        $new[$name] = $one;
                    }
                }
            }
        }

        foreach($new as &$lines) $lines = implode("\t",$lines);
        file_put_contents($file,implode("\n",$new));

        sem_release($this->lockid);
    }
    private function getCookie($url){
        $url_info = parse_url($url);
        $file = $this->config->cookie_store_path.'/'.$url_info['host'].'.cookie';
        $old = array();
        if(file_exists($file)){
            return ' -j "'.$file.'"';
        }
        return '';
    }

    public function getContents($row, &$wait_time){
        $url     = $row[0];
        $ticket  = $row[1];
        $referer = $row[2];

        $now_utime   = microtime(1);
        $site_path   = parse_url($url);
        $site_config = $this->sites[$site_path['host']];
        $_wait_time  = array();

        $query_sql   = 'SELECT `num`,`last_time` FROM ongoing WHERE `type`=? AND `type_id`=?';
        $query_sth   = $this->db->prepare($query_sql);

        shuffle($this->config->connects);
        foreach($this->config->connects as $connect_config){
            $connect_key = $this->getSign($connect_config);
            $site_key    = $this->getSign($connect_config, $site_path['host']);

            $rs = $query_sth->execute(array('connect',$connect_key));
            $connect_info = $query_sth->fetch(\PDO::FETCH_NUM);
            if($connect_info[0] >= $connect_config['max_connect']){ // 此连接已达到最大数
                #echo "\n连接{$connect_config['host']}已达到最大数\n";
                $_wait_time[] = $this->config->wait_usleep_time;
                continue;
            }
            $test_time = $connect_info[1] + $connect_config['get_gap']/1000000;
            if($now_utime < $test_time){ // 两次请求时间相隔太短
                #echo "\n连接{$connect_config['host']}两次请求时间相隔太短\n";
                $_wait_time[] = $connect_config['get_gap'];
                continue;
            }

            $rs = $query_sth->execute(array('site',$site_key));
            $site_info = $query_sth->fetch(\PDO::FETCH_NUM);
            if($site_info[0] >= $site_config['max_connect']){ // 此连接下的这个网站已达最大数
                #echo "\n在连接{$connect_config['host']}，网站{$site_path['host']}已达最大数\n";
                $_wait_time[] = $this->config->wait_usleep_time;
                continue;
            }
            $test_time = $site_info[1] + $site_config['get_gap']/1000000;
            if($now_utime < $test_time){ // 两次请求时间相隔太短
                #echo "\n在连接{$connect_config['host']}下，网站{$site_path['host']}两次请求时间相隔太短\n";
                $_wait_time[] = $site_config['get_gap'];
                continue;
            }

            $this->processing($connect_config, $url);
            $this->getContent($connect_config, $url, $ticket, $referer);
            return;
        }
        $wait_time = $_wait_time ? min($_wait_time) : $this->config->wait_usleep_time;
        #sleep(1);
        #echo "\n将等待 $wait_time 重试\n";
    }

    private function getContent($connect, $url, $ticket, $referer){
        if(empty($referer)){
            $referer = preg_replace('#(http://[^/]+)#','\1',$url);
        }
        $key = $connect['type'].'://'.$connect['host'];
        $extends = $this->header ? implode(' ',$this->header) : '';
        $cookies = $this->getCookie($url);
        $command = 'curl ';
        switch($connect['type']){
            case 'local':
                if($connect['host']!='127.0.0.1') $command .= ' --interface '.$connect['host'];
            break;
            case 'http':
            case 'socks4':
            case 'socks5':
                $type   = $connect['type'];
                $option = ($type=='http') ? ' -x ' : (" --$type ");
                $param  = explode('@',$connect['host']);
                if(isset($param[1])){
                    $host  = $option.$param[1];
                    $user  = ' -U '.$param[0];
                }else{
                    $host  = $option.$param[0];
                    $user  = '';
                }
                $command .= $user.$host;
            break;
        }
        $command .= " $url -i -k --connect-timeout 5 -m 5 --compressed -s -c - -A '{$this->agent_str}' -e $referer".$extends.$cookies;
        $command .= " | php {$this->receiver} {$this->config_file} $key $url $ticket >> {$this->config->error_log} 2>&1 &";
        #$command = "php {$this->receiver} {$this->config_file} $key $url $ticket >> {$this->config->error_log} 2>&1 &";
        #echo "\n $command.\n";

        $time   = date('m-d H:i:s');
        echo "\n[$ticket] [$time] [$key] $url";

        exec($command);
        $this->total++;
    }

    public function addUrls($urls,$ticket=0,$referer=null,$retry=false){
        $now_time = time();
        sem_acquire($this->lockid);
        $select_sql = 'SELECT `last_time` FROM `urls` WHERE `url`=?';
        $select_sth = $this->db->prepare($select_sql);

        $insert_sql = 'INSERT INTO `urls`(`url`,`ticket`,`referer`,`last_time`) ';
        $insert_sql.= 'VALUES(?,?,?,1) ON DUPLICATE KEY UPDATE `ticket`=VALUES(`ticket`),`last_time`=1';

        $insert_sth = $this->db->prepare($insert_sql);
        if($retry){
            foreach($urls as $url) $insert_sth->execute(array($url,$ticket,$referer));
        }else{
            foreach($urls as $url){
                $select_sth->execute(array($url));
                if($row = $select_sth->fetch(\PDO::FETCH_NUM)){   // 已存在数据库中
                    $url_info  = parse_url($url);
                    $site_info = $this->sites[$url_info['host']];
                    if($site_info['update_gap']){                 // 有设置抓取时间间隙
                        if($row[0] + $site_info['update_gap'] < $now_time){
                            $insert_sth->execute(array($url,$ticket,$referer));
                        }
                    }
                }else{
                    $insert_sth->execute(array($url,$ticket,$referer));
                }
            }
        }

        sem_release($this->lockid);
    }
    public function getUrls(){
        sem_acquire($this->lockid);
        $rs = $this->db->query('SELECT `url`,`ticket`,`referer` FROM `urls` WHERE `last_time`=1');
        sem_release($this->lockid);

        $urls = array();
        while($row = $rs->fetch(\PDO::FETCH_NUM)) $urls[] = $row;

        return $urls;
    }
    public function isProcessing(){
        sem_acquire($this->lockid);
        $rs = $this->db->query('SELECT `last_time` FROM `urls` WHERE `last_time`>0 AND `last_time`<127 LIMIT 1');
        $count = $rs->fetch(\PDO::FETCH_NUM);
        sem_release($this->lockid);
        return $count[0] > 0;
    }
    private function processing($connect, $url){
        $site_key    = $this->getSign($connect,$url);
        $connect_key = $this->getSign($connect);
        $now_time    = microtime(1);

        sem_acquire($this->lockid);
        $query_sql = 'UPDATE `ongoing` SET `num`=`num`+1, `last_time`=? WHERE `type`="site" AND `type_id`=?';
        $query_sth = $this->db->prepare($query_sql);
        $query_sth->execute(array($now_time, $site_key));

        $query_sql = 'UPDATE `ongoing` SET `num`=`num`+1, `last_time`=?  WHERE `type`="connect" AND `type_id`=?';
        $query_sth = $this->db->prepare($query_sql);
        $query_sth->execute(array($now_time, $connect_key));

        $query_sql = 'UPDATE `urls` SET `last_time`=2 WHERE `url`=?';
        $query_sth = $this->db->prepare($query_sql);
        $query_sth->execute(array($url));
        sem_release($this->lockid);
    }
    public function processed($connect_str, $url){
        $path_info   = parse_url($url);
        $site_key    = md5($connect_str.'|'.$path_info['host'],1);
        $connect_key = md5($connect_str,1);
        $now_time    = microtime(1);

        sem_acquire($this->lockid);
        $query_sql = 'UPDATE `ongoing` SET `num`=`num`-1, `last_time`=? WHERE `type`="site" AND `type_id`=?';
        $query_sth = $this->db->prepare($query_sql);
        $query_sth->execute(array($now_time, $site_key));

        $query_sql = 'UPDATE `ongoing` SET `num`=`num`-1, `last_time`=? WHERE `type`="connect" AND `type_id`=?';
        $query_sth = $this->db->prepare($query_sql);
        $query_sth->execute(array($now_time, $connect_key));

        $query_sql = 'UPDATE `urls` SET `last_time`=? WHERE `url`=?';
        $query_sth = $this->db->prepare($query_sql);
        if(!$this->sites[$path_info['host']]['update_gap']) $now_time = 0;
        $query_sth->execute(array($now_time,$url));
        sem_release($this->lockid);
    }
}
